Hive学习笔记(二) - UDF&UDTF

• UDF - 输入单列、输出单行单列
• UDTF - 输入单列、输出多行多列

1 UDF

1.1 准备数据

2017-12-20 11:15:37     test.***.user.act   {"app_platform":"554","channel_id":"554","inner_version":"554","model":"554","user_id":"i481781463"}    {"act":["UrlFree"],"act_id":["673"],"act_level":["None"],"act_num":"1","res_pos":["1"],"res_type":"ClientPopup","user_need":["FindBook","MoneyRunningOut"],"user_status":"ReadLocalBook"}
2017-12-20 11:15:38     test.***.user.act   {"app_platform":"554","channel_id":"554","inner_version":"554","model":"554","user_id":"i1078924274"}   {"act":["UrlFree"],"act_id":["673"],"act_level":["None"],"act_num":"1","res_pos":["1"],"res_type":"ClientPopup","user_need":["FindBook","NoMoneyToRead"],"user_status":"Unread"}
2017-12-20 11:15:39     test.***.user.act   {"app_platform":"554","channel_id":"554","inner_version":"554","model":"554","user_id":"i43843095"}     {"act":[],"act_id":[],"act_level":[],"act_num":"0","res_pos":[],"res_type":"None","user_need":["Other"],"user_status":"Other"}

1.2 建表、导入数据

# 创建日志表
create table activity_log(time string, topic string, common_json string, bussiness_json string)
row format delimited
fields terminated by '\t';

# load 数据
load data inpath '/Users/lixiwei-mac/app/hive/my_script/activity_log_data.log' intoo table activity_log;

1.3 编写UDF

package hive.udf;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;

import java.util.HashMap;
import java.util.Map;

/**
 * @author NikoBelic
 * @create 2017/12/20 11:51
 */
public class ActivityUserUDF extends UDF
{
    private static Map<String, String> map = new HashMap<>();


    public Map<String, String> evaluate(final Text s)
    {
        JSONObject commonJson = JSON.parseObject(s.toString());
        commonJson.entrySet().forEach(entry -> map.put(entry.getKey(), entry.getValue().toString()));
        map.put("remark", "卧槽牛逼啊老铁");
        return map;
    }
}

打成jar包

1.4 引入Hive

add JAR /Users/lixiwei-mac/app/hive/my_script/my_udf.jar; 
create temporary function trans_common_json as 'hive.udf.ActivityUserUDF';

1.5 测试

select trans_common_json(t.common_json) from activity_log t;

+---------------------------------------------------------------------------------------------------------------------------+--+
|                                                            _c0                                                            |
+---------------------------------------------------------------------------------------------------------------------------+--+
| {"app_platform":"554","inner_version":"554","user_id":"i481781463","model":"554","remark":"卧槽牛逼啊老铁","channel_id":"554"}   |
| {"app_platform":"554","inner_version":"554","user_id":"i1078924274","model":"554","remark":"卧槽牛逼啊老铁","channel_id":"554"}  |
| {"app_platform":"554","inner_version":"554","user_id":"i43843095","model":"554","remark":"卧槽牛逼啊老铁","channel_id":"554"}    |
+---------------------------------------------------------------------------------------------------------------------------+--+

2 UDTF

2.1 步骤

基本和UDF一样，只是编写的代码不同。

测试数据

John Smith  
John and Ann White  
Ted Green  
Dorothy

建外部表（需要导入到HDFS）

CREATE EXTERNAL TABLE people (name string)  
ROW FORMAT DELIMITED FIELDS   
    TERMINATED BY '\t'   
    ESCAPED BY ''   
    LINES TERMINATED BY '\n'  
STORED AS TEXTFILE   
LOCATION '/user/matthew/people';

2.2 编写UDTF

package hive.udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author NikoBelic
 * @create 2017/12/20 16:55
 */
public class NameParserGenericUDTF extends GenericUDTF
{

    private PrimitiveObjectInspector stringOI = null;

    @Override
    public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {

        if (args.length != 1) {
            throw new UDFArgumentException("NameParserGenericUDTF() takes exactly one argument");
        }

        if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE
                && ((PrimitiveObjectInspector) args[0]).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING) {
            throw new UDFArgumentException("NameParserGenericUDTF() takes a string as a parameter");
        }

        // 输入格式（inspectors）
        stringOI = (PrimitiveObjectInspector) args[0];

        // 输出格式（inspectors） -- 有两个属性的对象
        List<String> fieldNames = new ArrayList<String>(2);
        List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>(2);
        fieldNames.add("name");
        fieldNames.add("surname");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    public ArrayList<Object[]> processInputRecord(String name){
        ArrayList<Object[]> result = new ArrayList<Object[]>();

        // 忽略null值与空值
        if (name == null || name.isEmpty()) {
            return result;
        }

        String[] tokens = name.split("\\s+");

        if (tokens.length == 2){
            result.add(new Object[] { tokens[0], tokens[1] });
        }else if (tokens.length == 4 && tokens[1].equalsIgnoreCase("and")){
            result.add(new Object[] { tokens[0], tokens[3] });
            result.add(new Object[] { tokens[2], tokens[3] });
        }
        result.forEach(res -> System.out.println(res[0] + "," + res[1]));
        System.out.println("===============");
        return result;
    }

    @Override
    public void process(Object[] record) throws HiveException
    {

        final String name = stringOI.getPrimitiveJavaObject(record[0]).toString();

        ArrayList<Object[]> results = processInputRecord(name);

        Iterator<Object[]> it = results.iterator();

        while (it.hasNext()){
            Object[] r = it.next();
            forward(r);
        }
    }

    @Override
    public void close() throws HiveException {
        // do nothing
    }

    public static void main(String[] args)
    {
        List<String> testNames = new ArrayList<>();
        testNames.add("Niko Belic");
        testNames.add("James And Jnoh Snow");
        testNames.add("Kanalina Pino");
        NameParserGenericUDTF thisObj = new NameParserGenericUDTF();
        testNames.forEach(name -> thisObj.processInputRecord(name));
    }


}

2.3 测试

0: jdbc:hive2://0.0.0.0:10000> select process_name(p.name) from people p;
+-------+----------+--+
| name  | surname  |
+-------+----------+--+
| John  | Smith    |
| John  | White    |
| Ann   | White    |
| Ted   | Green    |
+-------+----------+--+

赏

晚饭加个鸡腿 ^.^

支付宝

微信